package com.fumasoft.dynamic.client;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.fumasoft.dynamic.config.RocketMqConfig;
import com.fumasoft.dynamic.listener.BillListener;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 消费者客户端.
 *
 * @author jingchen
 */
@Slf4j
@Configuration
public class ConsumerClient {

    @Autowired
    private RocketMqConfig rocketMqConfig;

    @Resource
    private BillListener billListener;

    /**
     * 单据消费者组.
     *
     * @return ConsumerBean
     */
    @Bean(name = "billConsumer", initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean billConsumer() {
        ConsumerBean consumerBean = new ConsumerBean();
        Properties properties = rocketMqConfig.getMqProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, rocketMqConfig.getBillGroup());
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        consumerBean.setProperties(properties);
        log.info("RocketMQ Consumer group  model: {}", properties.getProperty(PropertyKeyConst.MessageModel));
        //订阅消息
        Map<Subscription, MessageListener> portraitSubscription = new HashMap<Subscription, MessageListener>();
        Subscription subscription = new Subscription();
        subscription.setTopic(rocketMqConfig.getBillTopic());
        subscription.setExpression(rocketMqConfig.getBillTag());
        portraitSubscription.put(subscription, billListener);
        consumerBean.setSubscriptionTable(portraitSubscription);
        return consumerBean;
    }
}
